package com.larry.spark.test

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


/**
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的ID
 * 3	session_id	String	Session的ID
 * 4	page_id	Long	某个页面的ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的ID
 * 8	click_product_id	Long	某一个商品的ID
 * 9	order_category_ids	String	一次订单中所有品类的ID集合
 * 10	order_product_ids	String	一次订单中所有商品的ID集合
 * 11	pay_category_ids	String	一次支付中所有品类的ID集合
 * 12	pay_product_ids	String	一次支付中所有商品的ID集合
 * 13	city_id	Long	城市 id
 *
 */
object Top10_3 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    val sc = new SparkContext(conf)

    val dataLine = sc.textFile("data/user_visit_action.txt")
    //缓存
//    dataLine.cache()

    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //点击，下单，支付
    //点击数据

    val acc = new TopAcc()
    sc.register(acc,"acc")

    dataLine.foreach(
      data => {
        val datas = data.split("_")
        if (datas(6) != "-1"){
          acc.add((datas(6),"click"))
        }else if (datas(8) != "null"){
          val ids = datas(8).split(",")
          ids.map(
            id => {
              acc.add((id,"order"))
            }
          )
        }else if (datas(10) != "null"){
          val ids = datas(8).split(",")
          ids.map(
            id => {
              acc.add((id,"pay"))
            }
          )
        }
      }
    )

    val value = acc.value
    val top = value.map(_._2).toList.sortWith(
      (l, r) => {
        if (l.click > r.click) {
          true
        } else if (l.click == r.click) {
          if (l.order > r.order) {
            true
          } else if (l.order == r.order) {
            l.pay > r.pay
          } else {
            false
          }
        } else {
          false
        }
      }
    ).take(10)



    //排序 取前十
    top.foreach(println)

    sc.stop()
  }

  //样例类
  case class Category(cid : String,var click : Int,var order : Int,var pay : Int)

  //使用累加器完成
  class TopAcc extends AccumulatorV2[(String,String),mutable.Map[String,Category]] {

    private val map = mutable.Map[String,Category]()

    override def isZero: Boolean = {
      map.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, Category]] = {
      new TopAcc()
    }

    override def reset(): Unit = {
      map.clear()
    }

    override def add(v: (String, String)): Unit = {
      val (id,acctype) = v
      val category = map.getOrElse(id, Category(id, 0, 0, 0))
      acctype match {
        case "click" => category.click += 1
        case "order" => category.order += 1
        case "pay" => category.pay += 1
      }

      map.update(id,category)

    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, Category]]): Unit = {
      other.value.foreach{
        case (id,occ) => {
          val category = map.getOrElse(id, Category(id, 0, 0, 0))
          category.click += occ.click
          category.order += occ.order
          category.pay += occ.pay

          map.update(id,category)
        }
      }
    }

    override def value: mutable.Map[String, Category] = map
  }

}
